package com.github.jiangxch.rpc.transport.netty;

import com.github.jiangxch.rpc.serialization.Serializer;
import com.github.jiangxch.rpc.transport.AbstractEndpoint;
import com.github.jiangxch.rpc.transport.Channel;
import com.github.jiangxch.rpc.transport.ChannelHandler;
import com.github.jiangxch.rpc.transport.Server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LineBasedFrameDecoder;

import java.net.InetSocketAddress;
import java.util.Set;

import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * @author: jiangxch
 * @date: 2021/2/20 11:11
 */
@Slf4j
public class NettyServer extends AbstractEndpoint implements Server {
    private InetSocketAddress bindAddress;
    private ChannelHandler channelHandler;
    @Getter
    @Setter
    private Set<Channel> clientChannels;
    private ChannelFuture channelFuture;
    private boolean isClosed = true;
    private NioEventLoopGroup bossLoopGroup;
    private NioEventLoopGroup workLoopGroup;

    public NettyServer(InetSocketAddress bindAddress, ChannelHandler channelHandler, Serializer serializer) {
        super(serializer);
        this.bindAddress = bindAddress;
        this.channelHandler = channelHandler;

    }

    private void initServer() {
        // 初始化主从线程
        NioEventLoopGroup bossLoopGroup = new NioEventLoopGroup();
        NioEventLoopGroup workLoopGroup = new NioEventLoopGroup();

        try {
            // 初始化服务启动类
            ServerBootstrap serverBootstrap = new ServerBootstrap();

            // 配置主从线程，配置通道处理器
            serverBootstrap.group(bossLoopGroup, workLoopGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        // 添加自定义的通道处理器
                        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
                        ch.pipeline().addLast(new NettyChannelInboundHandler(channelHandler,NettyServer.this));
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true);
    
            channelFuture = serverBootstrap.bind(bindAddress.getPort());
            channelFuture.syncUninterruptibly();
            log.info("rpc socket server start success,bind address={}",bindAddress);
        } catch (Exception e) {
            log.error("rpc socket server start exception,", e);
        }
    }

    @Override
    public InetSocketAddress getLocalSocketAddress() {
        return bindAddress;
    }
    
    @Override
    public void start() {
        this.initServer();
        isClosed = false;
    }
    
    @Override
    public void close() {
        if (isClosed) {
            throw new RuntimeException("the netty server has closed");
        }
        // 关闭会阻塞
        channelFuture.channel().closeFuture().addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (future.isSuccess()) {
                    isClosed = true;
                    bossLoopGroup.shutdownGracefully();
                    workLoopGroup.shutdownGracefully();
                }
            }
        });
    }
    
    @Override
    public boolean isClosed() {
        return isClosed;
    }
}
